package com.yanggu.flink.datastream_api.window

import com.yanggu.flink.datastream_api.pojo.SensorReading
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.configuration.{Configuration, RestOptions}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * window function定义了要对窗口中收集的数据如何进行计算, 将WindowedStream => DataStream
 * 主要分为两类增量聚合函数和全窗口函数
 *
 * 增量聚合函数(incremental aggregation function)
 * 窗口内的数据来到就执行增量聚合函数, 保持一个简单的状态, 可以有效减少内存的占用。典型的有reduce方法和ReduceFunction以及aggregate方法AggregateFunction
 * ReduceFunction和AggregateFunction都是无状态的增量聚合函数, 不能使用状态
 * 需要注意的是ReduceFunction和AggregateFunction都不能是RichFunction, 可以查看WindowedStream的reduce的方法注释说明
 *
 * 全窗口函数(full window function)
 * 先把属于该窗口的所有数据收集起来, 等到计算的时候一起计算, 可以完成非常复杂的逻辑, 缺点是内存占用较大。典型的是process方法和ProcessWindowFunction
 *
 * 具有增量聚合的ProcessFunction
 * aggregate(AggregateFunction, ProcessWindowFunction)和reduce(ReduceFunction, ProcessWindowFunction)
 * 先执行增量聚合方法, 然后将增加聚合的结果进行响应输出
 *
 */
object WindowFunctionDemo {

  def main(args: Array[String]): Unit = {
    //创建本地执行环境, 并且拥有WebUi和设置端口
    val config = new Configuration()
    config.setInteger(RestOptions.PORT.key(), 8080)
    val environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config)
    environment.setParallelism(1)

    environment
      .socketTextStream("localhost", 9000)
      .map(value => {
        val arr = value.split(",")
        SensorReading(arr(0), arr(1).toLong * 1000, arr(2).toDouble)
      })
      //设置事件时间升序的Watermark
      .assignAscendingTimestamps(_.timestamp)
      //根据id进行分区
      .keyBy(_.id)
      .window(SlidingEventTimeWindows.of(Time.minutes(1L), Time.seconds(5L)))
      .reduce(new MyReduceFunction)
      .print("result ")

    environment.execute("WindowFunctionDemo Job")
  }

}

class MyReduceFunction extends ReduceFunction[SensorReading] {

  /**
   * 需要重写的核心方法
   * 逻辑获取最新数据的时间戳, 同时获取传感器温度的较小值
   * 这个方法只会在有两条以上的数据才会执行, 第一个数据来临时不会执行
   *
   * @param value1 中间聚合的值
   * @param value2 后续的值
   * @return
   */
  override def reduce(value1: SensorReading, value2: SensorReading): SensorReading = {
    SensorReading(value2.id, value2.timestamp, value1.temperature.min(value2.temperature))
  }

}
